package org.apache.rocketmq.client.impl;


import org.apache.rocketmq.client.ClientConfig;
import org.apache.rocketmq.client.Validators;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.attribute.AttributeParser;
import org.apache.rocketmq.common.namesrv.DefaultTopAddressing;
import org.apache.rocketmq.common.namesrv.NameServerUpdateCallback;
import org.apache.rocketmq.common.namesrv.TopAddressing;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.exception.RemotingConnectException;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.netty.NettyClientConfig;
import org.apache.rocketmq.remoting.netty.NettyRemotingClient;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.header.CreateTopicRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.DeleteTopicFromNamesrvRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.DeleteTopicRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.namesrv.GetKVConfigRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.namesrv.GetKVConfigResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.namesrv.GetRouteInfoRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.namesrv.PutKVConfigRequestHeader;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;

import java.util.Arrays;
import java.util.List;

/* @方法描述：控制台客户端的api对象，该对象内部持有了Netty构建的客户端，负责客户端与服务器的网络通信
 * 在第六版本代码中，该类也是最简化的
 */
public class MQClientAPIImpl implements NameServerUpdateCallback {

    private final static Logger log = LoggerFactory.getLogger(MQClientAPIImpl.class);


    static {
        System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
    }

    //  网络通信客户端
    private final RemotingClient remotingClient;


    //客户端请求处理器，专门处理服务端主动发送给客户端的请求，在第六版本代码中，根本没有实现这个请求处理器
    private final ClientRemotingProcessor clientRemotingProcessor;

    //Namesrv地址
    private String nameSrvAddr = null;


    private ClientConfig clientConfig;


    //Namesrv寻址器
    private final TopAddressing topAddressing;

    //构造方法
    public MQClientAPIImpl(final NettyClientConfig nettyClientConfig,
                           final ClientRemotingProcessor clientRemotingProcessor,
                           RPCHook rpcHook, final ClientConfig clientConfig) {
        this(nettyClientConfig, clientRemotingProcessor, rpcHook, clientConfig, null);
    }

    //构造方法
    public MQClientAPIImpl(final NettyClientConfig nettyClientConfig, final ClientRemotingProcessor clientRemotingProcessor, RPCHook rpcHook, final ClientConfig clientConfig, final ChannelEventListener channelEventListener) {
        this.clientConfig = clientConfig;
        //在这里创建Namesrv寻址器对象
        topAddressing = new DefaultTopAddressing(MixAll.getWSAddr(), clientConfig.getUnitName());
        //注册地址变更后的回调方法，当前类实现了NameServerUpdateCallback接口，所以当前类的对象就是一个回调对象
        //当Namesrv地址发生变更时，当前对象的onNameServerAddressChange()方法就会被回调
        topAddressing.registerChangeCallBack(this);
        //创建Netty客户端对象
        this.remotingClient = new NettyRemotingClient(nettyClientConfig, channelEventListener);
        this.clientRemotingProcessor = clientRemotingProcessor;
        //注册RPC回调方法
        this.remotingClient.registerRPCHook(rpcHook);
    }

    //启动Netty客户端的方法
    public void start() {
        this.remotingClient.start();
    }

    //关闭Netty客户端的方法
    public void shutdown() {
        this.remotingClient.shutdown();
    }

    //得到客户端保存的Namesrv地址列表的方法
    public List<String> getNameServerAddressList() {
        return this.remotingClient.getNameServerAddressList();
    }


    //得到Netty客户端的方法
    public RemotingClient getRemotingClient() {
        return remotingClient;
    }

    //使用Namesrv寻址器获取Namesrv地址的方法
    public String fetchNameServerAddr() {
        try {
            String addrs = this.topAddressing.fetchNSAddr();
            if (!UtilAll.isBlank(addrs)) {
                if (!addrs.equals(this.nameSrvAddr)) {
                    log.info("name server address changed, old=" + this.nameSrvAddr + ", new=" + addrs);
                    //地址发生变化则在这里更新最新的Namesrv地址到客户端中
                    this.updateNameServerAddressList(addrs);
                    this.nameSrvAddr = addrs;
                    return nameSrvAddr;
                }
            }
        } catch (Exception e) {
            log.error("fetchNameServerAddr Exception", e);
        }
        return nameSrvAddr;
    }


    //Namesrv地址列表发生变化后要回调的方法
    @Override
    public String onNameServerAddressChange(String namesrvAddress) {
        if (namesrvAddress != null) {
            if (!namesrvAddress.equals(this.nameSrvAddr)) {
                log.info("name server address changed, old=" + this.nameSrvAddr + ", new=" + namesrvAddress);
                this.updateNameServerAddressList(namesrvAddress);
                this.nameSrvAddr = namesrvAddress;
                return nameSrvAddr;
            }
        }
        return nameSrvAddr;
    }


    //更新最新的Namesrv地址到Netty客户端中的方法
    public void updateNameServerAddressList(final String addrs) {
        String[] addrArray = addrs.split(";");
        List<String> list = Arrays.asList(addrArray);
        this.remotingClient.updateNameServerAddressList(list);
    }

    /**
     * @Author jiaxiangen
     * @Description:
     * @Date 2025/6/16 16:21
     * @Param [namespace, key, value, timeoutMillis]
     * @方法描述：在指定的Broker节点上创建主题的方法
     */
    public void createTopic(final String addr, final String defaultTopic, final TopicConfig topicConfig,
                            final long timeoutMillis)
            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
        //校验要创建的主题信息的合法性
        Validators.checkTopicConfig(topicConfig);
        //创建主题请求头对象
        CreateTopicRequestHeader requestHeader = new CreateTopicRequestHeader();
        //设置主题名称
        requestHeader.setTopic(topicConfig.getTopicName());
        //设置默认的主题名称
        requestHeader.setDefaultTopic(defaultTopic);
        //设置读队列数量
        requestHeader.setReadQueueNums(topicConfig.getReadQueueNums());
        //设置写队列数量
        requestHeader.setWriteQueueNums(topicConfig.getWriteQueueNums());
        //设置主题权限
        requestHeader.setPerm(topicConfig.getPerm());
        //设置主题过滤类型
        requestHeader.setTopicFilterType(topicConfig.getTopicFilterType().name());
        //设置主题系统标志
        requestHeader.setTopicSysFlag(topicConfig.getTopicSysFlag());
        //设置主题是否支持顺序消息
        requestHeader.setOrder(topicConfig.isOrder());
        //设置主题扩展属性
        requestHeader.setAttributes(AttributeParser.parseToString(topicConfig.getAttributes()));
        RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_TOPIC, requestHeader);
        RemotingCommand remotingCommand = this.remotingClient.invokeSync(addr, requestCommand, timeoutMillis);
        switch (remotingCommand.getCode()) {
            case ResponseCode.SUCCESS:
                return;
            default:
                break;

        }
        throw new MQClientException("Unknown response code " + remotingCommand.getCode(), null);
    }

    /*
     * @方法描述：从指定的Broker节点上删除主题的方法
     */
    public void deleteTopicInBroker(final String addr, final String topic, final long timeoutMillis)
            throws RemotingException, InterruptedException, MQClientException {
        DeleteTopicRequestHeader requestHeader = new DeleteTopicRequestHeader();
        requestHeader.setTopic(topic);
        RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.DELETE_TOPIC_IN_BROKER, requestHeader);
        RemotingCommand remotingCommand = this.remotingClient.invokeSync(addr, requestCommand, timeoutMillis);
        assert remotingCommand != null;
        switch (remotingCommand.getCode()) {
            case ResponseCode.SUCCESS:
                return;
            default:
                break;

        }
        throw new MQClientException("Unknown response code " + remotingCommand.getCode(), null);
    }

    /*
     * @方法描述：从指定的NameServer节点上删除主题的方法，这个方法对端功能，也就是Namesrv接收并处理DeleteTopicFromNamesrvRequestHeader请求的功能，会在第七版本代码实现
     */
    public void deleteTopicInNameServer(final String addr, final String clusterName, final String topic,
                                        final long timeoutMillis)
            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
        DeleteTopicFromNamesrvRequestHeader requestHeader = new DeleteTopicFromNamesrvRequestHeader();
        requestHeader.setTopic(topic);
        requestHeader.setClusterName(clusterName);
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.DELETE_TOPIC_IN_NAMESRV, requestHeader);
        RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
        assert response != null;
        switch (response.getCode()) {
            case ResponseCode.SUCCESS: {
                return;
            }
            default:
                break;
        }

        throw new MQClientException(response.getCode(), response.getRemark());
    }

    //以下两个方法暂时不添加注释，等后面的版本代码实现对应功能再添加详细注释

    public void putKVConfigValue(final String namespace, final String key, final String value, final long timeoutMillis)
            throws RemotingException, MQClientException, InterruptedException {
        PutKVConfigRequestHeader requestHeader = new PutKVConfigRequestHeader();
        requestHeader.setNamespace(namespace);
        requestHeader.setKey(key);
        requestHeader.setValue(value);

        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PUT_KV_CONFIG, requestHeader);

        List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
        if (nameServerAddressList != null) {
            RemotingCommand errResponse = null;
            for (String namesrvAddr : nameServerAddressList) {
                RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMillis);
                assert response != null;
                switch (response.getCode()) {
                    case ResponseCode.SUCCESS: {
                        break;
                    }
                    default:
                        errResponse = response;
                }
            }

            if (errResponse != null) {
                throw new MQClientException(errResponse.getCode(), errResponse.getRemark());
            }
        }
    }


    public String getKVConfigValue(final String namespace, final String key, final long timeoutMillis)
            throws RemotingException, MQClientException, InterruptedException {
        GetKVConfigRequestHeader requestHeader = new GetKVConfigRequestHeader();
        requestHeader.setNamespace(namespace);
        requestHeader.setKey(key);

        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_KV_CONFIG, requestHeader);
        RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis);
        assert response != null;
        switch (response.getCode()) {
            case ResponseCode.SUCCESS: {
                GetKVConfigResponseHeader responseHeader =
                        (GetKVConfigResponseHeader) response.decodeCommandCustomHeader(GetKVConfigResponseHeader.class);
                return responseHeader.getValue();
            }
            default:
                break;
        }

        throw new MQClientException(response.getCode(), response.getRemark());
    }

    public TopicRouteData getTopicRouteInfoFromNameServer(final String topic, final long timeoutMillis)
            throws RemotingException, MQClientException, InterruptedException {
        return getTopicRouteInfoFromNameServer(topic, timeoutMillis, true);
    }
    public TopicRouteData getTopicRouteInfoFromNameServer(final String topic, final long timeoutMillis,
                                                          boolean allowTopicNotExist) throws MQClientException, InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
        //创建获取主题路由信息的请求头对象
        GetRouteInfoRequestHeader requestHeader = new GetRouteInfoRequestHeader();
        //设置主题名称
        requestHeader.setTopic(topic);
        //创建请求对象
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ROUTEINFO_BY_TOPIC, requestHeader);
        //发送请求到Namesrv节点，并且得到响应
        RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis);
        assert response != null;
        switch (response.getCode()) {
            case ResponseCode.TOPIC_NOT_EXIST: {
                if (allowTopicNotExist) {
                    log.warn("get Topic [{}] RouteInfoFromNameServer is not exist value", topic);
                }

                break;
            }
            case ResponseCode.SUCCESS: {
                byte[] body = response.getBody();
                if (body != null) {
                    //响应成功则解码响应体，返回TopicRouteData对象
                    return TopicRouteData.decode(body, TopicRouteData.class);
                }
            }
            default:
                break;
        }

        throw new MQClientException(response.getCode(), response.getRemark());
    }
    public ClusterInfo getBrokerClusterInfo(long timeoutMillis) throws InterruptedException, RemotingTimeoutException,
            RemotingSendRequestException, RemotingConnectException, MQBrokerException {

        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_CLUSTER_INFO, null);
        RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis);

        assert response != null;
        switch (response.getCode()) {
            case ResponseCode.SUCCESS: {
                return ClusterInfo.decode(response.getBody(), ClusterInfo.class);
            }
            default:
                break;
        }
        throw new MQBrokerException(response.getCode(), response.getRemark());

    }
}